package com.zm.xiaohashu.relation.biz.consumer;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.zm.utils.DateUtils;
import com.zm.xiaohashu.relation.biz.constant.MQConstants;
import com.zm.xiaohashu.relation.biz.constant.RedisKeyConstants;
import com.zm.xiaohashu.relation.biz.domain.Fans;
import com.zm.xiaohashu.relation.biz.domain.Following;
import com.zm.xiaohashu.relation.biz.enums.FollowUnfollowTypeEnum;
import com.zm.xiaohashu.relation.biz.mapper.FansMapper;
import com.zm.xiaohashu.relation.biz.mapper.FollowingMapper;
import com.zm.xiaohashu.relation.biz.model.dto.CountFollowUnfollowMqDTO;
import com.zm.xiaohashu.relation.biz.model.dto.FollowUserMqDTO;
import com.zm.xiaohashu.relation.biz.model.dto.UnfollowUserMqDTO;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Component;
import org.springframework.transaction.support.TransactionTemplate;

import java.time.LocalDateTime;
import java.util.Collections;
import java.util.Objects;

@Component
@RocketMQMessageListener(consumerGroup = "xiaohashu_group" + MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW, // Group 组
        topic = MQConstants.TOPIC_FOLLOW_OR_UNFOLLOW, // 消费的 Topic 主题
        consumeMode = ConsumeMode.ORDERLY
)
@Slf4j
public class FollowUnfollowConsumer implements RocketMQListener<Message> {

    @Resource
    private FollowingMapper followingMapper;
    @Resource
    private FansMapper fansMapper;
    @Resource
    private TransactionTemplate transactionTemplate;
    // 每秒创建 5000 个令牌
    private RateLimiter rateLimiter = RateLimiter.create(5000);
    @Resource
    private RedisTemplate<String, Object> redisTemplate;
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void onMessage(Message message) {
        // 流量削峰：通过获取令牌，如果没有令牌可用，将阻塞，直到获得
        rateLimiter.acquire();
        // 消息体
        String bodyJsonStr = new String(message.getBody());
        // 标签
        String tags = message.getTags();

        log.info("==> FollowUnfollowConsumer 消费了消息 {}, tags: {}", bodyJsonStr, tags);

        if (Objects.equals(tags, MQConstants.TAG_FOLLOW)) {
            handleFollowTagMessage(bodyJsonStr);
        } else if (Objects.equals(tags, MQConstants.TAG_UNFOLLOW)) {
            handleUnfollowTagMessage(bodyJsonStr);
        }
    }

    /**
     * 处理关注消息
     *
     * @param bodyJsonStr
     */
    private void handleFollowTagMessage(String bodyJsonStr) {
        FollowUserMqDTO followUserMqDTO = JSONUtil.toBean(bodyJsonStr, FollowUserMqDTO.class);
        if (Objects.isNull(followUserMqDTO)) return;
        Long userId = followUserMqDTO.getUserId();
        Long followUserId = followUserMqDTO.getFollowUserId();
        LocalDateTime createTime = followUserMqDTO.getCreateTime();
        boolean result = transactionTemplate.execute(status -> {
            try {
                Following following = Following.builder()
                        .userId(userId)
                        .followingUserId(followUserId)
                        .createTime(createTime)
                        .build();
                int count = followingMapper.insert(following);
                if (count > 0) {
                    Fans fans = Fans.builder()
                            .userId(followUserId)
                            .fansUserId(userId)
                            .createTime(createTime).build();
                    fansMapper.insert(fans);
                }
                return true;
            } catch (Exception e) {
                status.setRollbackOnly();
                return false;
            }
        });

        //更新Redis中用户被关注用户的ZSet
        if (result) {
            // 时间戳
            /**
             * 这里得的lua脚本就是去做判断，存在的话就直接修改，不存在就不做处理
             */
            long timestamp = DateUtils.localDateTime2Timestamp(createTime);
            DefaultRedisScript<Long> script = new DefaultRedisScript<>();
            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/follow/follow_check_and_update_fans_zset.lua")));
            script.setResultType(Long.class);
            String fansRedisKey = RedisKeyConstants.buildUserFansKey(followUserId);
            redisTemplate.execute(script, Collections.singletonList(fansRedisKey), userId, timestamp);
        }
        //mq计数
        // 构建消息体 DTO
        CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder()
                .userId(userId)
                .targetUserId(followUserId)
                .type(FollowUnfollowTypeEnum.FOLLOW.getCode()) // 关注
                .build();

        senMessage(countFollowUnfollowMqDTO);
    }


    /**
     * 处理取消关注消息
     *
     * @param bodyJsonStr
     */
    private void handleUnfollowTagMessage(String bodyJsonStr) {
        if (StrUtil.isBlank(bodyJsonStr)) return;
        UnfollowUserMqDTO unfollowUserMqDTO = JSONUtil.toBean(bodyJsonStr, UnfollowUserMqDTO.class);
        Long unfollowUserId = unfollowUserMqDTO.getUnfollowUserId();
        Long userId = unfollowUserMqDTO.getUserId();
        LocalDateTime createTime = unfollowUserMqDTO.getCreateTime();
        boolean isSuccess = transactionTemplate.execute(status -> {
            try {
                LambdaQueryWrapper<Following> follwingLuw = new LambdaQueryWrapper<>();
                follwingLuw.eq(Following::getUserId, userId);
                follwingLuw.eq(Following::getFollowingUserId, unfollowUserId);
                followingMapper.delete(follwingLuw);
                LambdaUpdateWrapper<Fans> fansLuw = new LambdaUpdateWrapper<>();
                fansLuw.eq(Fans::getUserId, unfollowUserId);
                fansLuw.eq(Fans::getFansUserId, userId);
                fansMapper.delete(fansLuw);
                return true;
            } catch (Exception e) {
                status.setRollbackOnly();
                return false;
            }
        });
        if (isSuccess) {
            //更新redis
            String fansKey = RedisKeyConstants.buildUserFansKey(unfollowUserId);
            DefaultRedisScript<Long> script = new DefaultRedisScript<>();
            script.setResultType(Long.class);
            script.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/unfollow/unfollow_check_and_update_fans_zset.lua")));
            redisTemplate.execute(script, Collections.singletonList(fansKey), userId);
        }
        //mq计数
        CountFollowUnfollowMqDTO countFollowUnfollowMqDTO = CountFollowUnfollowMqDTO.builder()
                .userId(unfollowUserId)
                .targetUserId(userId)
                .type(FollowUnfollowTypeEnum.UNFOLLOW.getCode()) // 关注
                .build();
        senMessage(countFollowUnfollowMqDTO);
    }


    private void senMessage(CountFollowUnfollowMqDTO countFollowUnfollowMqDTO) {
        org.springframework.messaging.Message<CountFollowUnfollowMqDTO> message = MessageBuilder.withPayload(countFollowUnfollowMqDTO).build();
        String destination1 = MQConstants.TOPIC_COUNT_FOLLOWING;
        rocketMQTemplate.asyncSend(destination1, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务：关注数入库】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务：关注数入库】MQ 发送异常: ", throwable);
            }
        });
        String destination2 = MQConstants.TOPIC_COUNT_FANS;
        rocketMQTemplate.asyncSend(destination2, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> 【计数服务：粉丝数入库】MQ 发送成功，SendResult: {}", sendResult);
            }

            @Override
            public void onException(Throwable throwable) {
                log.error("==> 【计数服务：粉丝数入库】MQ 发送异常: ", throwable);
            }
        });

    }

}
